/*
 *  Queen Mary University of London MOM System
 *  
 *  Copyright Queen Mary University of London
 *  
 *  Authors:
 *  Bob Chew (bob.chew@elec.qmul.ac.uk)
 *  Beatriz Viñal Murciano (beavimu@gmail.com)
 *  John Bigham (john.bigham@elec.qmul.ac.uk)
 *  
 *  
 *  Portions of this software uses Apache Qpid
 *    http://qpid.apache.org
 *  
 *  Apache Qpid license:
 * --------------------------------------------------------------
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 * -------------------------------------------------------------
 * 
 */

package brokerBottleneckController;

import java.io.FileInputStream;
import java.util.Arrays;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Properties;
import java.util.Set;

import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.TopicSubscriber;

import momControlClient.MOMControlClient;

import org.apache.qpid.client.TopicPublisherAdapter;
import org.apache.qpid.client.message.JMSTextMessage;

/**
 * Receives messages from the detectors periodically. With that information, run
 * the bottleneck classifier. (Currently only a simple threshold test.) When a
 * broker is bottlenecked, we send a message to the overlay manager.
 */
public class Controller extends MOMControlClient implements MessageListener {
	private boolean __DEBUG = false;

	/** Machine in which the controller is running. */
	private final String controllerIP;
	/** Topic used to communicate with the detector. */
	private static final String TOPIC_CONTROLLER = "brokerBottleneck.controller";
	/** Subscriber to listen to detectors. */
	private TopicSubscriber subscriber;
	/** Number of parts that there must be in a received message. */
	private static final int NUM_PARTS_MSG = 2;

	/** Topic used to communicate with the overlay manager. */
	private static final String TOPIC_OM = "brokerFailure.overlayManager";
	/** Publisher to send messages to the overlay manager. */
	private TopicPublisherAdapter omPublisher;

	/** List of machines in which a broker is running. */
	private static Set<String> brokers;

	/**
	 * Getter for the variable 'brokers'.
	 * 
	 * @return variable 'brokers'
	 */
	public static Set<String> getBrokers() {
		return brokers;
	}

	/**
	 * Setter for the variable 'brokers'.
	 * 
	 * @param brokers
	 *            new value of the variable 'brokers'
	 */
	public static void setBrokers(final Set<String> brokers) {
		Controller.brokers = brokers;
	}

	/**
	 * List of machines in which a broker is running and their state. State is
	 * an int from WORST_STATE (worst) to BEST_STATE (best)
	 */
	private static HashMap<String, Integer> brokersState;

	/**
	 * Getter for the variable 'brokersState'.
	 * 
	 * @return variable 'brokersState'
	 */
	public static HashMap<String, Integer> getBrokersState() {
		return brokersState;
	}

	/**
	 * Setter for the variable 'brokersState'.
	 * 
	 * @param brokersState
	 *            new value of the variable 'brokersState'
	 */
	public static void setBrokersState(
			final HashMap<String, Integer> brokersState) {
		Controller.brokersState = brokersState;
	}

	/** Best state of a broker. */
	private static final int BEST_STATE = 10;
	/** Worst state of a broker. */
	private static final int WORST_STATE = 0;

	/**
	 * List of machines in which a broker is running and the states in which
	 * they detect each one of the other brokers.
	 */
	private HashMap<String, HashMap<String, Integer>> brokersDetection;

	/**
	 * Constructor.
	 * 
	 * @param controllerIP
	 *            IP address in which the controller broker is running (its own
	 *            IP address)
	 * @param brokerAddresses
	 *            list of brokers in the system mapped to the ports they are
	 *            listening at
	 */
	public Controller(final String controllerIP,
			final HashMap<String, Integer> brokerAddresses) {

		super(); // assign ID

		this.controllerIP = controllerIP;

		setBrokers(brokerAddresses.keySet());

		// at the beginning, assume that all brokers are off
		setBrokersState(new HashMap<String, Integer>());
		for (String broker : getBrokers()) {
			getBrokersState().put(broker, WORST_STATE);
		}

		// at the beginning, assume that all detectors detect all brokers off
		brokersDetection = new HashMap<String, HashMap<String, Integer>>();
		for (String broker : getBrokers()) {
			HashMap<String, Integer> detection = new HashMap<String, Integer>();
			for (String brkr : getBrokers()) {
				detection.put(brkr, WORST_STATE);
			}
			brokersDetection.put(broker, detection);
		}

		// start a connection to read messages from the detectors and to send
		// messages to the overlay manager
		try {

			// create and start a connection and a session
			openConnAndSess(this.controllerIP);

			// create a subscriber for listening to the detectors
			subscriber = createSubscriber(TOPIC_CONTROLLER);
			// set the subscriber to listen to messages
			subscriber.setMessageListener(this);

			// create a publisher to send messages to the overlay manager
			omPublisher = createPublisher(TOPIC_OM);

			System.out.println("Controller waiting for messages in "
					+ this.controllerIP);

		} catch (Exception e) {
			System.err.println(e.getMessage());
		}

	}

	/**
	 * Get the property names of the topics that are being published to the
	 * given broker.
	 * 
	 * @param broker
	 *            to look for topics published to it
	 * @return set of prop names of topics published to the given broker
	 */
	private Set<String> getTopicsPubToBroker(final String broker) {
		Set<String> topics = new HashSet<String>();

		// Read broker-topic mapping file (from OM)
		String PUB_ASSIG_FILE_NAME = "brokerPublishers.xml";
		Properties pubAssignments = new Properties();
		try {
			FileInputStream fis = new FileInputStream(PUB_ASSIG_FILE_NAME);
			pubAssignments.loadFromXML(fis);
			if (__DEBUG)
				pubAssignments.list(System.out);
		} catch (Exception e) {
			System.err.println("");
			e.printStackTrace();
		}

		Set<String> propPubNames = pubAssignments.stringPropertyNames();

		for (String propPubName : propPubNames) {
			// if a publisher was using that broker
			if (((String) pubAssignments.get(propPubName)).equals(broker)) {
				topics.add(propPubName);
			}
		}

		if (__DEBUG)
			System.out.println(Arrays.deepToString(topics.toArray())
					.replaceAll(",", ",\n"));

		return topics;
	}

	private void updateOM(String bottleneckedBroker) throws JMSException {
		String text = "broker failed:" + bottleneckedBroker;
		System.out.println("message to Overlay Manager: " + "\"" + text + "\"");
		sendMessage(omPublisher, text);

		System.out.println("Should have switched brokers!");
	}

	/**
	 * Read the detector, broker and state reported in a received message and
	 * report the change.
	 * 
	 * This method is invoked asynchronously when a message is received.
	 * 
	 * @param msg
	 *            received message
	 */
	// @Override
	public void onMessage(final Message msg) {
		try {
			JMSTextMessage message = (JMSTextMessage) msg;
			String text = message.getText();
			String[] parts = text.split(":");
			if (parts.length == NUM_PARTS_MSG) {
				System.out.println("> " + text);
				updateOM(parts[1]);
			} else {
				throw new Exception("Controller: Bad format of the message");
			}
		} catch (Exception e) {
			System.err
					.println("Controller: Error receiving or parsing message.");
			e.printStackTrace();
		}
	}

}
